/*
 * This file is a part of qloudgen-esb. 
 * You can redistribute qloudgen-esb and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-esb is distributed WITHOUT ANY WARRANTY. 
 * 
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/.
 *  
 * Copyright (C) 2014 Qloudgen Technology
*/

package com.qloudgen.csb;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.log4j.Logger;
import org.zeromq.ZContext;

import kafka.consumer.ConsumerConfig;

public class Receiver implements Runnable { 
	private String topic;
	private String groupId;
	private String zkConnect;
	private int numThreads;
	private int bufferSize;
	private ZContext ctx;
	private String conn;
    private int timeout;
    private int retry;
    private int mode = 0;
    private ErrorHandler errHandler;
    private final String id = CSBUtils.sockId();
    
	private ExecutorService executor;
	private ConsumerConfig config;
	
    private static Logger LOG = Logger.getLogger(Receiver.class.getName());

    public Receiver(String topic, String groupId, String zkConnect, 
    	int numThreads, int bufferSize, ZContext ctx, String conn,
    		int timeout, int retry, int mode, ErrorHandler errHandler) {
        this.topic = topic;
        this.groupId = groupId;
        this.zkConnect = zkConnect;
        this.numThreads = numThreads;
        this.bufferSize = bufferSize;
        this.ctx = ctx;
        this.conn = conn;
        this.timeout = timeout;
        this.retry = retry;
        this.mode = mode;
        this.errHandler = errHandler;
    }

    public void run() {    	
    	config = createConsumerConfig();       
 
    	Future<?>[] tasks = new Future<?>[numThreads];
        executor = Executors.newFixedThreadPool(numThreads);
        for (int i = 0; i < numThreads; i++) {
        	if (mode == 0)
        		tasks[i] = executor.submit(new ZmqConsumerS
        				(topic, config, bufferSize, ctx, conn, 
        						timeout, retry, errHandler));            
        	else 
        		tasks[i] = executor.submit(new ZmqConsumerP
        				(topic, config, bufferSize, ctx, conn, 
        						timeout, retry, errHandler));            
        }     
        
        while (true) {
            try {
                Thread.sleep(CSB.RECEIVER_CHECK_INTERVAL);
                checkTasks(tasks);
            } catch (Exception e) {
            	LOG.error(id + ": task check error, exit loop...");
                break;
            }
        }              
    }
    
    private void checkTasks(Future<?>[] tasks) throws Exception {
    	for (int i = 0; i < numThreads; i++) {
    		if (tasks[i] == null
                || tasks[i].isDone()
                || tasks[i].isCancelled())
    		{
    			if (mode == 0)
    				tasks[i] = executor.submit(new ZmqConsumerS
    					(topic, config, bufferSize, ctx, conn, 
    							timeout, retry, errHandler));
    			else
    				tasks[i] = executor.submit(new ZmqConsumerP
    					(topic, config, bufferSize, ctx, conn, 
    							timeout, retry, errHandler));
    			LOG.warn(id + ": consumer stopped, restarting...");
    		}
    	}
    }
    
    private ConsumerConfig createConsumerConfig() {
    	Properties props = new Properties();
    	FileInputStream fis = null;
		try {
			fis = new FileInputStream("conf/csbconsumer.properties");
			props.load(fis);
		} catch (IOException e) {			
		} finally {
			if (fis != null)
				try {
					fis.close();
				} catch (IOException e) {}
		}
		
        props.put("zookeeper.connect", zkConnect);
        props.put("group.id", groupId);
        props.put("auto.commit.enable", "false");
        props.put("auto.offset.reset", "largest");
        props.put("consumer.timeout.ms", "-1");
        props.put("client.id", Receiver.class.getName());
        
        return new ConsumerConfig(props);
    }  
}

